package com.aspire.kafkaproducer.config;


import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * kafka生产者配置类
 *
 * @author kingSon
 *
 * @version 1.0.0
 *
 */
@Configuration
@EnableKafka
public class KafkaProducerConfiguration {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducerConfiguration.class);

    @Autowired
    private KafkaConf kafkaConf;

    private static KafkaProducerConfiguration producer = null;

    /**
     * 设置kafka连接工厂
     *
     * @return Map<String, Object>
     */
    private Map<String, Object> producerConfigs() {
        final Map<String, Object> props = new ConcurrentHashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaConf.getBootstrapServers());
        props.put(ProducerConfig.RETRIES_CONFIG, this.kafkaConf.getRetries());
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, this.kafkaConf.getBatchSize());
        props.put(ProducerConfig.LINGER_MS_CONFIG, this.kafkaConf.getLingerMs());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        LOGGER.info(props.toString());
        return props;
    }

    private ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<String, String>(this.producerConfigs());
    }

    /**
     * kafka相关api操作入口
     *
     * @return KafkaTemplate<String, String>
     */
    @Bean(name = "kafkaTemplate")
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> kafkaTemplate = null;
        try {
            kafkaTemplate = new KafkaTemplate<String, String>(this.producerFactory());
        } catch (Exception e) {
            LOGGER.error("创建KafkaTemplate时出错", e);
            return null;
        }
        return kafkaTemplate;
    }
}